package com.zx.mes.hyl.rocket.mq.producer;

import com.zx.mes.hyl.rocket.mq.util.RocketUtil;
import com.zx.mes.hyl.rocket.mq.yml.RocketMqProperties;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;

/**
 * rocket mq producer
 * <p>
 * example:
 *
 * @author 华云龙
 * @date 2018-12-10
 */
@Log4j2
public class TransactionProducer {

    /**
     * rocket mq 下面的生产者
     */
    private TransactionMQProducer producer;
    /**
     * 从外面设置rocket mq properties 对象
     */
    private RocketMqProperties rocketMqProperties;

    private TransactionListener transactionListener;

    /**
     * 设置成功则返回true,失败则返回false
     *
     * @param producerGroupName 生产者组
     * @return true/false
     */
    public void producerConfig(String producerGroupName) {
        /// 检查配置
        RocketMqProperties.RocketMq rocketMq = RocketUtil.checkRocketMqConfig(producerGroupName, rocketMqProperties);
        if (rocketMq != null && StringUtils.isNotBlank(rocketMq.getNameServer())) {
            /// 由于生产者组可能有多个,所以是map形式
            this.producer = new TransactionMQProducer(producerGroupName);
            /// rocket mq 所需要的线程配置
            RocketMqProperties.RocketMqThreadPool rocketMqThreadPool = rocketMqProperties.getRocketMqThreadPool();
            ExecutorService executorService = new ThreadPoolExecutor(rocketMqThreadPool.getCorePoolSize(), rocketMqThreadPool.getMaximumPoolSize(), rocketMqThreadPool.getMaximumPoolSize(), rocketMqThreadPool.getUnit(),
                    new ArrayBlockingQueue<>(rocketMqThreadPool.getArrayBlockingQueueCapacity()), r -> {
                Thread thread = new Thread(r);
                thread.setName(producerGroupName + "-check-thread");
                return thread;
            });
            this.producer.setTransactionListener(transactionListener);
            this.producer.setExecutorService(executorService);
            this.producer.setNamesrvAddr(rocketMq.getNameServer());
        } else {
            log.error("生产者组:["+ producerGroupName +"],对应的配置信息错误,请检查!");
        }

    }

    public void producerConfig(String producerGroupName,RocketMqProperties rocketMqProperties,TransactionListener transactionListener){
        this.rocketMqProperties = rocketMqProperties;
        this.transactionListener = transactionListener;
        this.producerConfig(producerGroupName);
    }

    /**
     * 发送消息
     *
     * @param message  消息体
     * @param argument 消息所带参数
     * @return 返回 事务消息状态
     */
    public TransactionSendResult sendMessage(Message message, Object argument) {
        this.start();
        TransactionSendResult sendResult = null;
        try {
            sendResult = this.producer.sendMessageInTransaction(message, argument);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return sendResult;
    }


    private void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    /**
     * 向外提供 transaction producer shutdown 接口
     */
    public void shutdown(){
        this.producer.shutdown();
    }

}
